package com.zdb.demo.flink.test;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;

public class DataStreamDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);  //设置时间分配器

        env.setParallelism(1);  //设置并行度
        env.getConfig().setAutoWatermarkInterval(9000);//每9秒发出一个watermark

        DataStream<String> text = env.socketTextStream("localhost", 9999);

        DataStream<Tuple3<String, Long, Integer>> counts = text.filter(new FilterClass()).map(new LineSplitter())
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple3<String, Long, Integer>>() {

                    private long currentMaxTimestamp = 0L;

                    private final long maxOutOfOrderness = 10000L;   //这个控制失序已经延迟的度量
                    //获取EventTime
                    @Override
                    public long extractTimestamp(Tuple3<String, Long, Integer> element, long previousElementTimestamp) {
                        long timestamp = element.f1;
                        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                        System.out.println(
                                "get timestamp is " + timestamp + " currentMaxTimestamp " + currentMaxTimestamp);
                        return timestamp;
                    }
                    //获取Watermark
                    @Override
                    public Watermark getCurrentWatermark() {
                        System.out.println("wall clock is " + System.currentTimeMillis() + " new watermark "
                                + (currentMaxTimestamp - maxOutOfOrderness));
                        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
                    }
                }).keyBy(0).timeWindow(Time.seconds(20))
                // .allowedLateness(Time.seconds(10))   
                .reduce(new ReduceFunction<Tuple3<String, Long, Integer>>() {
                            @Override public Tuple3<String, Long, Integer> reduce(
                                    Tuple3<String, Long, Integer> t1,
                                    Tuple3<String, Long, Integer> t2) throws Exception {
                                return new Tuple3<>(t1.f0, t2.f1, t1.f2 + t2.f2);
                            }
                        },
                        new WindowFunction<Tuple3<String, Long, Integer>, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() {
                            @Override public void apply(Tuple tuple, TimeWindow window,
                                    Iterable<Tuple3<String, Long, Integer>> input,
                                    Collector<Tuple3<String, Long, Integer>> out) throws Exception {
                                long end = window.getEnd();
                                final Tuple3<String, Long, Integer>[]
                                        tuple3 =
                                        new Tuple3[] { null };
                                input.forEach(t3 -> {
                                    if (tuple3[0] == null) {
                                        tuple3[0] = t3;
                                    } else {
                                        tuple3[0] = new Tuple3<>(tuple3[0].f0, t3.f1, tuple3[0].f2 + t3.f2);
                                    }
                                });

                                out.collect(new Tuple3<>(tuple3[0].f0, end, tuple3[0].f2));
                            }
                        });

        counts.print();
        env.execute("Window WordCount");

    }

//    自定义获取timesStamp
//    private static class MyTimestamp extends AscendingTimestampExtractor<Tuple3<String, Long, Integer>> {
//
//        private static final long serialVersionUID = 1L;
//
//        public long extractAscendingTimestamp(Tuple3<String, Long, Integer> element) {
//
//            return element.f1;
//        }
//
//    }

    // 构造出element以及它的event time.然后把次数赋值为1
    public static final class LineSplitter implements
            MapFunction<String, Tuple3<String, Long, Integer>> {

        @Override
        public Tuple3<String, Long, Integer> map(String value) throws Exception {
            String[] tokens = value.toLowerCase().split("\\W+");

            long eventtime = Long.parseLong(tokens[1]) * 1000L;

            return new Tuple3<String, Long, Integer>(tokens[0], eventtime, 1);
        }
    }

    //过滤掉为null和whitespace的字符串
    public static final class FilterClass implements FilterFunction<String> {

        @Override
        public boolean filter(String value) throws Exception {

            if (StringUtils.isNullOrWhitespaceOnly(value)) {
                return false;
            } else {
                return true;
            }
        }

    }
}